﻿using System;
using System.Collections;
using System.Collections.Generic;
using System.ComponentModel;
using System.Linq;
using System.Reflection;
using System.Runtime.CompilerServices;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
using Microsoft.Extensions.Logging;
using Workstation.Collections;
using Workstation.ServiceModel.Ua.Channels;

namespace Workstation.ServiceModel.Ua
{
    /// <summary>
    /// 订阅以从 OPC UA 服务器接收数据更改和事件的基类。
    /// </summary>
    public abstract class SubscriptionBase : INotifyPropertyChanged, INotifyDataErrorInfo, ISetDataErrorInfo
    {
        private readonly ActionBlock<PublishResponse> _actionBlock;
        private readonly IProgress<CommunicationState> _progress;
        private readonly ILogger? _logger;
        private readonly UaApplication _application;
        private volatile bool _isPublishing;
        private volatile ClientSessionChannel? _innerChannel;
        private volatile uint _subscriptionId;
        private readonly ErrorsContainer<string> _errors;
        private PropertyChangedEventHandler? _propertyChanged;
        private string? _endpointUrl;
        private double _publishingInterval = ClientSessionChannel.DefaultPublishingInterval;
        private uint _keepAliveCount = ClientSessionChannel.DefaultKeepaliveCount;
        private uint _lifetimeCount;
        private readonly MonitoredItemBaseCollection _monitoredItems = new();
        private CommunicationState _state = CommunicationState.Created;
        private volatile TaskCompletionSource<bool> _whenSubscribed;
        private volatile TaskCompletionSource<bool> __whenUnsubscribed;
        private CancellationTokenSource? _stateMachineCts;
        private Task? _stateMachineTask;
        private event Action<DataValue, string?>? _onDataValue;
        /// <summary>
        /// 初始化新的实例<see cref="SubscriptionBase"/> class.
        /// </summary>
        public SubscriptionBase() : this(UaApplication.Current)
        {

        }

        /// <summary>
        /// 初始化新的实例<see cref="SubscriptionBase"/> class.
        /// </summary>
        /// <param name="application">The UaApplication.</param>
        private SubscriptionBase(UaApplication? application)
        {
            _application = application ?? throw new ArgumentNullException(nameof(application));
            _application.Completion.ContinueWith(t => _stateMachineCts?.Cancel());
            _logger = _application.LoggerFactory?.CreateLogger(GetType());
            _errors = new ErrorsContainer<string>(p => ErrorsChanged?.Invoke(this, new DataErrorsChangedEventArgs(p)));
            _progress = new Progress<CommunicationState>(s => State = s);
            _propertyChanged += OnPropertyChanged!;
            _whenSubscribed = new TaskCompletionSource<bool>();
            __whenUnsubscribed = new TaskCompletionSource<bool>();
            __whenUnsubscribed.TrySetResult(true);
            _onDataValue += OnDataValueChanged;
            // 注册要在 UI 线程上运行的操作（如果有）。
            if (SynchronizationContext.Current is not null)
            {
                //订阅通知回调
                _actionBlock = new ActionBlock<PublishResponse>(OnPublishResponse, new ExecutionDataflowBlockOptions
                {
                    SingleProducerConstrained = true,
                    //作为任务调度器来确保回调在UI线程上执行。
                    TaskScheduler = TaskScheduler.FromCurrentSynchronizationContext()
                });
            }
            else
            {
                _actionBlock = new ActionBlock<PublishResponse>(OnPublishResponse,
                                                                new ExecutionDataflowBlockOptions { SingleProducerConstrained = true });
            }
        }

        public void InitializeComponent(SubscriptionAttribute subscriptionAttribute, Dictionary<string, string> propNodeIdMaps)
        {
            if (subscriptionAttribute is null)
            {
                throw new ArgumentNullException(nameof(subscriptionAttribute));
            }
            if (propNodeIdMaps is null)
            {
                throw new ArgumentNullException(nameof(propNodeIdMaps));
            }

            var typeInfo = GetType().GetTypeInfo();

            _endpointUrl = subscriptionAttribute.EndpointUrl;
            _publishingInterval = subscriptionAttribute.PublishingInterval;
            _keepAliveCount = subscriptionAttribute.KeepAliveCount;
            _lifetimeCount = subscriptionAttribute.LifetimeCount;

            // 获取 [MonitoredItem] 特性标记.typeInfo.DeclaredProperties
            foreach (var propertyInfo in this.GetType().GetProperties())
            {
                string? nodeId;

                var monitoredItemAttribute = propertyInfo.GetCustomAttribute<MonitoredItemAttribute>();

                if (monitoredItemAttribute is null)
                {
                    continue;
                }
                else
                {
                    if (monitoredItemAttribute.TagName is null || string.IsNullOrEmpty(monitoredItemAttribute.TagName))
                    {
                        continue;
                    }
                    propNodeIdMaps.TryGetValue(monitoredItemAttribute.TagName, out nodeId);

                    if (string.IsNullOrEmpty(nodeId))
                    {
                        continue;
                    }
                }

                MonitoringFilter? filter = null;
                if (monitoredItemAttribute.AttributeId == AttributeIds.Value && (monitoredItemAttribute.DataChangeTrigger != DataChangeTrigger.StatusValue || monitoredItemAttribute.DeadbandType != DeadbandType.None))
                {
                    //过滤器
                    filter = new DataChangeFilter()
                    {
                        Trigger = monitoredItemAttribute.DataChangeTrigger,
                        DeadbandType = (uint)monitoredItemAttribute.DeadbandType,
                        DeadbandValue = monitoredItemAttribute.DeadbandValue
                    };
                }

                var propType = propertyInfo.PropertyType;

                if (propType == typeof(DataValue))
                {
                    _monitoredItems.Add(new DataValueMonitoredItem(
                        target: this,
                        property: propertyInfo,
                        nodeId: ExpandedNodeId.Parse(nodeId),
                        indexRange: monitoredItemAttribute.IndexRange,
                        attributeId: monitoredItemAttribute.AttributeId,
                        samplingInterval: monitoredItemAttribute.SamplingInterval,
                        filter: filter,
                        queueSize: monitoredItemAttribute.QueueSize,
                        discardOldest: monitoredItemAttribute.DiscardOldest));
                    continue;
                }

                if (propType == typeof(BaseEvent) || propType.GetTypeInfo().IsSubclassOf(typeof(BaseEvent)))
                {
                    _monitoredItems.Add(new EventMonitoredItem(
                        target: this,
                        property: propertyInfo,
                        nodeId: ExpandedNodeId.Parse(nodeId),
                        indexRange: monitoredItemAttribute.IndexRange,
                        attributeId: monitoredItemAttribute.AttributeId,
                        samplingInterval: monitoredItemAttribute.SamplingInterval,
                        filter: new EventFilter() { SelectClauses = EventHelper.GetSelectClauses(propType) },
                        queueSize: monitoredItemAttribute.QueueSize,
                        discardOldest: monitoredItemAttribute.DiscardOldest));
                    continue;
                }

                if (propType == typeof(ObservableQueue<DataValue>))
                {
                    _monitoredItems.Add(new DataValueQueueMonitoredItem(
                        target: this,
                        property: propertyInfo,
                        nodeId: ExpandedNodeId.Parse(nodeId),
                        indexRange: monitoredItemAttribute.IndexRange,
                        attributeId: monitoredItemAttribute.AttributeId,
                        samplingInterval: monitoredItemAttribute.SamplingInterval,
                        filter: filter,
                        queueSize: monitoredItemAttribute.QueueSize,
                        discardOldest: monitoredItemAttribute.DiscardOldest));
                    continue;
                }
                //是否泛型且泛型类型的基础定义是ObservableQueue
                if (propType.IsConstructedGenericType && propType.GetGenericTypeDefinition() == typeof(ObservableQueue<>))
                {
                    var elemType = propType.GenericTypeArguments[0];
                    if (elemType == typeof(BaseEvent) || elemType.GetTypeInfo().IsSubclassOf(typeof(BaseEvent)))
                    {
                        _monitoredItems.Add((MonitoredItemBase)Activator.CreateInstance(
                        typeof(EventQueueMonitoredItem<>).MakeGenericType(elemType),
                        this,
                        propertyInfo,
                        ExpandedNodeId.Parse(nodeId),
                        monitoredItemAttribute.AttributeId,
                        monitoredItemAttribute.IndexRange,
                        MonitoringMode.Reporting,
                        monitoredItemAttribute.SamplingInterval,
                        new EventFilter() { SelectClauses = EventHelper.GetSelectClauses(elemType) },
                        monitoredItemAttribute.QueueSize,
                        monitoredItemAttribute.DiscardOldest)!);
                        continue;
                    }
                }
                //var typeArray = propType.GetGenericArguments();
                //if (typeArray.Length > 0)
                //{
                //    propType = typeArray[0];
                //}
                _monitoredItems.Add((MonitoredItemBase)Activator.CreateInstance(
                    typeof(ValueMonitoredItem<>).MakeGenericType(propType),
                    this,
                    propertyInfo,
                    ExpandedNodeId.Parse(nodeId),
                    monitoredItemAttribute.AttributeId,
                    monitoredItemAttribute.IndexRange,
                    MonitoringMode.Reporting,
                    monitoredItemAttribute.SamplingInterval,
                    filter,
                    monitoredItemAttribute.QueueSize,
                    monitoredItemAttribute.DiscardOldest)!);
            }
            _stateMachineCts = new CancellationTokenSource();
            _stateMachineTask = Task.Run(() => StateMachineAsync(_stateMachineCts.Token));
        }

        /// <inheritdoc/>
        public event PropertyChangedEventHandler? PropertyChanged
        {
            add
            {
                var flag = _propertyChanged?.GetInvocationList().Length == 1;
                _propertyChanged += value;
                if (flag)
                {
                    __whenUnsubscribed = new TaskCompletionSource<bool>();
                    _whenSubscribed.TrySetResult(true);
                }
            }

            remove
            {
                _propertyChanged -= value;
                if (_propertyChanged?.GetInvocationList().Length == 1)
                {
                    _whenSubscribed = new TaskCompletionSource<bool>();
                    __whenUnsubscribed.TrySetResult(true);
                }
            }
        }

        /// <summary>
        /// Gets the <see cref="CommunicationState"/>.
        /// </summary>
        public CommunicationState State
        {
            get { return _state; }
            private set { SetProperty(ref _state, value); }
        }

        /// <summary>
        /// Gets the current subscription Id.
        /// </summary>
        public uint SubscriptionId => _state == CommunicationState.Opened ? _subscriptionId : 0u;

        /// <summary>
        /// Requests a Refresh of all Conditions.
        /// </summary>
        /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
        public async Task<StatusCode> ConditionRefreshAsync()
        {
            if (this.State != CommunicationState.Opened)
            {
                return StatusCodes.BadServerNotConnected;
            }

            return await this.InnerChannel.ConditionRefreshAsync(this.SubscriptionId);
        }

        /// <summary>
        /// Acknowledges a condition.
        /// </summary>
        /// <param name="condition">an AcknowledgeableCondition.</param>
        /// <param name="comment">a comment.</param>
        /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
        public async Task<StatusCode> AcknowledgeAsync(AcknowledgeableCondition condition, LocalizedText? comment = null)
        {
            if (condition == null)
            {
                throw new ArgumentNullException(nameof(condition));
            }

            if (State != CommunicationState.Opened)
            {
                return StatusCodes.BadServerNotConnected;
            }

            return await InnerChannel.AcknowledgeAsync(condition, comment);
        }

        /// <summary>
        /// Confirms a condition.
        /// </summary>
        /// <param name="condition">an AcknowledgeableCondition.</param>
        /// <param name="comment">a comment.</param>
        /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
        public async Task<StatusCode> ConfirmAsync(AcknowledgeableCondition condition, LocalizedText? comment = null)
        {
            if (condition == null)
            {
                throw new ArgumentNullException(nameof(condition));
            }

            if (this.State != CommunicationState.Opened)
            {
                return StatusCodes.BadServerNotConnected;
            }

            return await this.InnerChannel.ConfirmAsync(condition, comment);
        }

        /// <summary>
        /// Gets the inner channel.
        /// </summary>
        protected ClientSessionChannel InnerChannel
        {
            get
            {
                if (_innerChannel == null)
                {
                    throw new ServiceResultException(StatusCodes.BadServerNotConnected);
                }

                return this._innerChannel;
            }
        }

        /// <summary>
        /// 设置属性值并通知侦听器属性值已更改。
        /// </summary>
        /// <typeparam name="T">Type of the property.</typeparam>
        /// <param name="oldValue">Reference to a storage field.</param>
        /// <param name="newValue">The new value.</param>
        /// <param name="propertyName">Name of the property used to notify listeners. This
        /// value is optional and can be provided automatically when invoked from compilers that
        /// support CallerMemberName.</param>
        /// <returns>True if the value changed, otherwise false.</returns>
        protected virtual bool SetProperty<T>(ref T oldValue, T newValue, [CallerMemberName] string? propertyName = null)
        {
            if (Equals(oldValue, newValue))
            {
                return false;
            }
            if (_isPublishing)
            {
                oldValue = newValue;
                NotifyPropertyChanged(propertyName);
            }
            else
            {
                if (newValue is CommunicationState)
                {
                    oldValue = newValue;
                    NotifyPropertyChanged(propertyName);
                    return true;
                }
                _onDataValue?.Invoke(new DataValue(newValue), propertyName);
            }
            return true;
        }

        protected virtual bool SetPropertyEx<T>(ref T oldValue, T newValue, [CallerMemberName] string? propertyName = null)
        {
            if (Equals(oldValue, newValue))
            {
                return false;
            }
            oldValue = newValue;
            NotifyPropertyChanged(propertyName);
            return true;
        }

        /// <summary>
        /// Notifies listeners that the property value has changed.
        /// </summary>
        /// <param name="propertyName">Name of the property used to notify listeners. This
        /// value is optional and can be provided automatically when invoked from compilers
        /// that support <see cref="T:System.Runtime.CompilerServices.CallerMemberNameAttribute" />.</param>
        protected virtual void NotifyPropertyChanged([CallerMemberName] string? propertyName = null)
        {
            this._propertyChanged?.Invoke(this, new PropertyChangedEventArgs(propertyName));
        }

        /// <summary>
        /// Occurs when the validation errors have changed for a property or entity.
        /// </summary>
        public event EventHandler<DataErrorsChangedEventArgs>? ErrorsChanged;

        /// <summary>
        /// Gets a value indicating whether the entity has validation errors.
        /// </summary>
        public bool HasErrors
        {
            get { return this._errors.HasErrors; }
        }

        /// <summary>
        /// Gets the validation errors for a specified property or for the entire entity.
        /// </summary>
        /// <param name="propertyName">The name of the property to retrieve validation errors for, or null or System.String.Empty to retrieve entity-level errors.</param>
        /// <returns>The validation errors for the property or entity.</returns>
        public IEnumerable GetErrors(string? propertyName)
        {
            return this._errors.GetErrors(propertyName);
        }

        /// <summary>
        /// Sets the validation errors for a specified property or for the entire entity.
        /// </summary>
        /// <param name="propertyName">The name of the property, or null or System.String.Empty to set entity-level errors.</param>
        /// <param name="errors">The validation errors for the property or entity.</param>
        void ISetDataErrorInfo.SetErrors(string propertyName, IEnumerable<string>? errors)
        {
            this._errors.SetErrors(propertyName, errors);
        }

        /// <summary>
        /// Handle PublishResponse message.
        /// </summary>
        /// <param name="publishResponse">The publish response.</param>
        private void OnPublishResponse(PublishResponse publishResponse)
        {
            _isPublishing = true;
            try
            {
                // 循环遍历所有通知
                if (publishResponse.NotificationMessage?.NotificationData == null)
                {
                    return;
                }

                foreach (var notificationData in publishResponse.NotificationMessage.NotificationData)
                {
                    // 如果数据更改。
                    if (notificationData is not null && notificationData is DataChangeNotification dcn)
                    {
                        if (dcn.MonitoredItems is not null)
                        {
                            foreach (var min in dcn.MonitoredItems)
                            {
                                if (min?.Value == null)
                                {
                                    _logger?.LogError($"One of the monitored item notifications is null");
                                    continue;
                                }

                                if (_monitoredItems.TryGetValueByClientId(min.ClientHandle, out var item))
                                {
                                    try
                                    {
                                        //通知
                                        item.Publish(min.Value);
                                    }
                                    catch (Exception ex)
                                    {
                                        _logger?.LogError($"Error publishing value for NodeId {item.NodeId}. {ex.Message}");
                                    }
                                }
                            }
                            continue;
                        }
                    }

                    // 如果事件。
                    var enl = notificationData as EventNotificationList;
                    if (enl?.Events != null)
                    {
                        foreach (var efl in enl.Events)
                        {
                            if (efl?.EventFields == null)
                            {
                                _logger?.LogError($"One of the event field list is null");
                                continue;
                            }

                            if (_monitoredItems.TryGetValueByClientId(efl.ClientHandle, out var item))
                            {
                                try
                                {
                                    item.Publish(efl.EventFields);
                                }
                                catch (Exception ex)
                                {
                                    _logger?.LogError($"Error publishing event for NodeId {item.NodeId}. {ex.Message}");
                                }
                            }
                        }
                    }
                }
            }
            finally
            {
                _isPublishing = false;
            }
        }

        /// <summary>
        /// Handles PropertyChanged event. If the property is associated with a MonitoredItem, writes the property value to the node of the server.
        /// </summary>
        /// <param name="sender">the sender.</param>
        /// <param name="e">the event.</param>
        private async void OnPropertyChanged(object sender, PropertyChangedEventArgs e)
        {
            if (this._isPublishing || string.IsNullOrEmpty(e.PropertyName))
            {
                return;
            }

            if (this._monitoredItems.TryGetValueByName(e.PropertyName, out var item))
            {
                if (item.TryGetValue(out var value))
                {
                    StatusCode statusCode;
                    try
                    {
                        var writeRequest = new WriteRequest
                        {
                            NodesToWrite = new[] { new WriteValue
                            {
                                NodeId = ExpandedNodeId.ToNodeId(item.NodeId, this.InnerChannel.NamespaceUris),
                                AttributeId = item.AttributeId,
                                IndexRange = item.IndexRange,
                                Value = value
                            } }
                        };
                        var writeResponse = await this.InnerChannel.WriteAsync(writeRequest).ConfigureAwait(false);
                        statusCode = writeResponse?.Results?[0] ?? StatusCodes.BadDataEncodingInvalid;
                    }
                    catch (ServiceResultException ex)
                    {
                        statusCode = ex.StatusCode;
                    }
                    catch (Exception)
                    {
                        statusCode = StatusCodes.BadServerNotConnected;
                    }

                    item.OnWriteResult(statusCode);
                    if (StatusCode.IsBad(statusCode))
                    {
                        this._logger?.LogError($"Error writing value for {item.NodeId}. {StatusCodes.GetDefaultMessage(statusCode)}");
                    }
                }
            }
        }

        private async void OnDataValueChanged(DataValue dataValue, string? propertyName)
        {
            if (this._isPublishing || string.IsNullOrEmpty(propertyName))
            {
                return;
            }

            if (this._monitoredItems.TryGetValueByName(propertyName!, out var item))
            {
                StatusCode statusCode;
                try
                {
                    var writeRequest = new WriteRequest
                    {
                        NodesToWrite = new[] { new WriteValue
                            {
                                NodeId = ExpandedNodeId.ToNodeId(item.NodeId, this.InnerChannel.NamespaceUris),
                                AttributeId = item.AttributeId,
                                IndexRange = item.IndexRange,
                                Value = dataValue
                            } }
                    };
                    var writeResponse = await this.InnerChannel.WriteAsync(writeRequest).ConfigureAwait(false);
                    statusCode = writeResponse?.Results?[0] ?? StatusCodes.BadDataEncodingInvalid;
                }
                catch (ServiceResultException ex)
                {
                    statusCode = ex.StatusCode;
                }
                catch (Exception)
                {
                    statusCode = StatusCodes.BadServerNotConnected;
                }

                item.OnWriteResult(statusCode);
                if (StatusCode.IsBad(statusCode))
                {
                    this._logger?.LogError($"Error writing value for {item.NodeId}. {StatusCodes.GetDefaultMessage(statusCode)}");
                }
            }
        }
        /// <summary>
        /// Signals the channel state is Closing.
        /// </summary>
        /// <param name="channel">The session channel. </param>
        /// <param name="token">A cancellation token. </param>
        /// <returns>A task.</returns>
        private async Task WhenChannelClosingAsync(ClientSessionChannel channel, CancellationToken token = default)
        {
            var tcs = new TaskCompletionSource<bool>();
            EventHandler handler = (o, e) =>
            {
                tcs.TrySetResult(true);
            };
            using (token.Register(state => ((TaskCompletionSource<bool>)state!).TrySetCanceled(), tcs, false))
            {
                try
                {
                    channel.Closing += handler;
                    if (channel.State == CommunicationState.Opened)
                    {
                        await tcs.Task;
                    }
                }
                finally
                {
                    channel.Closing -= handler;
                }
            }
        }

        /// <summary>
        /// 状态机管理订阅的状态。
        /// </summary>
        /// <param name="token">A cancellation token.</param>
        /// <returns>A task.</returns>
        private async Task StateMachineAsync(CancellationToken token = default)
        {
            while (!token.IsCancellationRequested)
            {
                await this._whenSubscribed.Task;

                this._progress.Report(CommunicationState.Opening);

                try
                {
                    if (this._endpointUrl is null)
                    {
                        throw new InvalidOperationException("The endpointUrl field must not be null. Please, use the Subscription attribute properly.");
                    }

                    // get a channel.
                    this._innerChannel = await this._application.GetChannelAsync(this._endpointUrl, token);
                    try
                    {
                        // create the subscription.
                        var subscriptionRequest = new CreateSubscriptionRequest
                        {
                            RequestedPublishingInterval = this._publishingInterval,
                            RequestedMaxKeepAliveCount = this._keepAliveCount,
                            RequestedLifetimeCount = Math.Max(this._lifetimeCount, 3 * this._keepAliveCount),
                            PublishingEnabled = true
                        };

                        var subscriptionResponse = await this._innerChannel.CreateSubscriptionAsync(subscriptionRequest).ConfigureAwait(false);

                        // link up the dataflow blocks
                        var id = this._subscriptionId = subscriptionResponse.SubscriptionId;
                        var linkToken = this._innerChannel.LinkTo(this._actionBlock, pr => pr.SubscriptionId == id);

                        try
                        {
                            // create the monitored items.
                            var items = this._monitoredItems.ToList();
                            if (items.Count > 0)
                            {
                                var requests = items.Select(m => new MonitoredItemCreateRequest { ItemToMonitor = new ReadValueId { NodeId = ExpandedNodeId.ToNodeId(m.NodeId, this.InnerChannel.NamespaceUris), AttributeId = m.AttributeId, IndexRange = m.IndexRange }, MonitoringMode = m.MonitoringMode, RequestedParameters = new MonitoringParameters { ClientHandle = m.ClientId, DiscardOldest = m.DiscardOldest, QueueSize = m.QueueSize, SamplingInterval = m.SamplingInterval, Filter = m.Filter } }).ToArray();

                                //split requests array to MaxMonitoredItemsPerCall chunks
                                int maxmonitoreditemspercall = 100;
                                MonitoredItemCreateRequest[] requests_chunk;
                                int chunk_size;
                                for (int i_chunk = 0; i_chunk < requests.Length; i_chunk += maxmonitoreditemspercall)
                                {
                                    chunk_size = Math.Min(maxmonitoreditemspercall, requests.Length - i_chunk);
                                    requests_chunk = new MonitoredItemCreateRequest[chunk_size];
                                    Array.Copy(requests, i_chunk, requests_chunk, 0, chunk_size);

                                    var itemsRequest = new CreateMonitoredItemsRequest
                                    {
                                        SubscriptionId = id,
                                        ItemsToCreate = requests_chunk,
                                    };
                                    var itemsResponse = await this._innerChannel.CreateMonitoredItemsAsync(itemsRequest);

                                    if (itemsResponse.Results is { } results)
                                    {
                                        for (int i = 0; i < results.Length; i++)
                                        {
                                            var item = items[i];
                                            var result = results[i];

                                            if (result is null)
                                            {
                                                this._logger?.LogError($"Error creating MonitoredItem for {item.NodeId}. The result is null.");
                                                continue;
                                            }

                                            item.OnCreateResult(result);
                                            if (StatusCode.IsBad(result.StatusCode))
                                            {
                                                this._logger?.LogError($"Error creating MonitoredItem for {item.NodeId}. {StatusCodes.GetDefaultMessage(result.StatusCode)}");
                                            }
                                        }
                                    }
                                }
                            }

                            this._progress.Report(CommunicationState.Opened);

                            // wait here until channel is closing, unsubscribed or token cancelled.
                            try
                            {
                                await Task.WhenAny(
                                    this.WhenChannelClosingAsync(this._innerChannel, token),
                                    this.__whenUnsubscribed.Task);
                            }
                            catch
                            {
                            }
                            finally
                            {
                                this._progress.Report(CommunicationState.Closing);
                            }
                        }
                        catch (Exception ex)
                        {
                            this._logger?.LogError($"Error creating MonitoredItems. {ex.Message}");
                            this._progress.Report(CommunicationState.Faulted);
                        }
                        finally
                        {
                            linkToken.Dispose();
                        }

                        if (this._innerChannel.State == CommunicationState.Opened)
                        {
                            try
                            {
                                // delete the subscription.
                                var deleteRequest = new DeleteSubscriptionsRequest
                                {
                                    SubscriptionIds = new uint[] { id }
                                };
                                await this._innerChannel.DeleteSubscriptionsAsync(deleteRequest);
                            }
                            catch (Exception ex)
                            {
                                this._logger?.LogError($"Error deleting subscription. {ex.Message}");
                                await Task.Delay(2000);
                            }
                        }

                        this._progress.Report(CommunicationState.Closed);
                    }
                    catch (Exception ex)
                    {
                        this._logger?.LogError($"Error creating subscription. {ex.Message}");
                        this._progress.Report(CommunicationState.Faulted);
                        await Task.Delay(2000);
                    }
                }
                catch (Exception ex)
                {
                    this._logger?.LogTrace($"Error getting channel. {ex.Message}");
                    this._progress.Report(CommunicationState.Faulted);
                    await Task.Delay(2000);
                }
            }
        }
    }
}
